package handler

import (
	"context"
	"gitee.com/easy_code/go-redis/cluster"
	"gitee.com/easy_code/go-redis/config"
	"gitee.com/easy_code/go-redis/database"
	databaseface "gitee.com/easy_code/go-redis/interface/database"
	"gitee.com/easy_code/go-redis/lib/logger"
	"gitee.com/easy_code/go-redis/lib/sync/atomic"
	"gitee.com/easy_code/go-redis/resp/connection"
	"gitee.com/easy_code/go-redis/resp/parser"
	"gitee.com/easy_code/go-redis/resp/reply"
	"io"
	"net"
	"strings"
	"sync"
)

var (
	unknownErrReplyBytes = []byte("-ERR unknown\r\n")
)

type RespHandler struct {
	activeConn sync.Map // 活跃的客户端
	db         databaseface.Database
	closing    atomic.Boolean
}

func MakeHandler() *RespHandler {
	var db databaseface.Database
	if config.Properties.Self != "" && len(config.Properties.Peers) > 0 { // 集群版
		db = cluster.MakeClusterDatabase()
	} else { // 单机版
		db = database.NewStandaloneDatabase()
	}
	return &RespHandler{
		db: db,
	}
}

// closeClient 关闭单个客户端
func (r *RespHandler) closeClient(client *connection.Connection) {
	_ = client.Close()
	r.db.AfterClientClose(client)
	r.activeConn.Delete(client)
}

// Handle 处理 TCP 连接
func (r *RespHandler) Handle(ctx context.Context, conn net.Conn) {
	// 关闭中
	if r.closing.Get() {
		// closing handler refuse new connection
		_ = conn.Close()
		return
	}

	// 记录连接
	client := connection.NewConn(conn)
	r.activeConn.Store(client, struct{}{})

	logger.Info("client connection")

	ch := parser.ParseStream(conn)
	for payload := range ch {
		// error
		if payload.Err != nil {
			// 客户端在做TCP挥手
			if payload.Err == io.EOF ||
				payload.Err == io.ErrUnexpectedEOF ||
				// 被关闭的连接
				strings.Contains(payload.Err.Error(), "use of closed network connection") {
				r.closeClient(client)
				logger.Info("connection closed：" + client.RemoteAddr().String())
				return
			}
			// 协议错误，发送错误信息给客户端
			errReply := reply.MakeErrReply(payload.Err.Error())
			err := client.Write(errReply.ToBytes())
			if err != nil { // 回写错误，tcp 中断
				r.closeClient(client)
				logger.Info("connection closed：" + client.RemoteAddr().String())
				return
			}
			continue
		}

		// exec
		if payload.Data == nil {
			continue
		}
		replyData, ok := payload.Data.(*reply.MultiBulkReply)
		if !ok {
			logger.Error("require multi bulk protocol")
			continue
		}
		result := r.db.Exec(client, replyData.Args)
		if result != nil {
			_ = client.Write(result.ToBytes())
		} else {
			// 未知错误
			_ = client.Write(unknownErrReplyBytes)
		}
	}
}

// Close 关闭整个handler（所有client）
func (r *RespHandler) Close() error {
	logger.Info("handler shutting down")
	r.closing.Set(true) // 设置连接已关闭，防止新的连接进入
	r.activeConn.Range( // 关闭全部活跃的连接
		func(key, value interface{}) bool {
			client := key.(*connection.Connection)
			_ = client.Close()
			return true
		})
	r.db.Close()
	return nil
}
